﻿using Microsoft.Hadoop.Client;
using Microsoft.Hadoop.MapReduce;
using Microsoft.Hadoop.WebClient.WebHCatClient;
using Microsoft.Hadoop.WebHDFS;
using Microsoft.Hadoop.WebHDFS.Adapters;
using Microsoft.Win32;
using Microsoft.WindowsAzure.Management.HDInsight;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.ComponentModel;
using System.ComponentModel.DataAnnotations;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Reflection;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Input;
using ViewMaker;
using ViewMaker.Core;
using ViewMaker.Core.Controls;

namespace HDInsightStudio
{

    public class MapReduceJobItem
    {
        public MapReduceJobItem(string filename)
        {
            this.Filename = filename;
        }

        public string Name
        {
            get { return Path.GetFileNameWithoutExtension(Filename); }
        }

        public string Filename { get; set; }

    }
    [View(typeof(MapReduceStudioView))]
    [ViewProperty(ViewControl.Properties.WindowHeight, 400)]
    [ViewProperty(ViewControl.Properties.WindowWidth, 600)]
    public class MapReduceStudio : JobViewModelBase
    {

        public MapReduceJobItem MapJob
        {
            get { return _MapJob; }
            set { _MapJob = value; OnPropertyChanged("MapJob"); }
        }
        private MapReduceJobItem _MapJob;

        public MapReduceJobItem ReduceJob
        {
            get { return _ReduceJob; }
            set { _ReduceJob = value; OnPropertyChanged("ReduceJob"); }
        }
        private MapReduceJobItem _ReduceJob;

        public MapReduceJobItem CombinerJob
        {
            get { return _CombinerJob; }
            set { _CombinerJob = value; OnPropertyChanged("CombinerJob"); }
        }
        private MapReduceJobItem _CombinerJob;

        public string Output
        {
            get { return _output; }
            set { _output = value; }
        }
        private string _output;

        public string Input
        {
            get { return _input; }
            set { _input = value; OnPropertyChanged("Input"); }
        }
        private string _input;


        public string Arguments
        {
            get { return _Arguments; }
            set { _Arguments = value; }
        }
        private string _Arguments;

        public string Defines
        {
            get { return _Defines; }
            set { _Defines = value; }
        }
        private string _Defines;

        public string Files
        {
            get { return _Files; }
            set { _Files = value; OnPropertyChanged("Files"); }
        }
        private string _Files;

        public bool CopyFiles
        {
            get { return _CopyFiles; }
            set { _CopyFiles = value; }
        }
        private bool _CopyFiles;

        public bool Busy
        {
            get { return _busy; }
            set { _busy = value; OnPropertyChanged(null); }
        }
        private bool _busy;

        public string StdErr
        {
            get { return _StdErr; }
            set { _StdErr = value; OnPropertyChanged("StdErr"); }
        }
        private string _StdErr;

        private string HomeFolder = "/MapReduceStudio";
        public MapReduceStudio():this(DateTime.Now.ToString("yyyyMMddHHmmss"))
        {
        }

        public MapReduceStudio(string jobId):base(jobId)
        {
            Output = HomeFolder + "/" + jobId + "/Output/";
            Status = HomeFolder + "/" + jobId + "/Status/";

        }

        public ICommand SelectMapCommand
        {
            get { return this.CreateCommand(SelectMap, (x) => !Busy); }
        }
        private void SelectMap()
        {
            var dialog = new OpenFileDialog();
            dialog.Filter = "*.exe|*.exe";
            if (dialog.ShowDialog() == true)
            {
                MapJob = new MapReduceJobItem(dialog.FileName);
            }
        }

        public ICommand SelectCombinerCommand
        {
            get { return this.CreateCommand(SelectCombiner, (x) => !Busy); }
        }
        private void SelectCombiner()
        {
            var dialog = new OpenFileDialog();
            dialog.Filter = "*.exe|*.exe";
            if (dialog.ShowDialog() == true)
            {
                CombinerJob = new MapReduceJobItem(dialog.FileName);
            }
        }


        public ICommand SelectReduceCommand
        {
            get { return this.CreateCommand(SelectReduce, (x) => !Busy); }
        }
        private void SelectReduce()
        {
            var dialog = new OpenFileDialog();
            dialog.Filter = "*.exe|*.exe";
            if (dialog.ShowDialog() == true)
            {
                ReduceJob = new MapReduceJobItem(dialog.FileName);
            }
        }

        public ICommand ExecuteMapReduceCommand
        {
            get { return this.CreateCommand(ExecuteMapReduce, (x) => !Busy); }
        }
        private void ExecuteMapReduce()
        {
            if (MapJob != null || ReduceJob != null)
            {
                BackgroundWorker worker = new BackgroundWorker();
                worker.DoWork += (s, e) =>
                {
                    Busy = true;

                    string mapper = null;
                    string reducer = null;
                    string combiner = null;
                    string mapTargetfilepath = null;
                    string reduceTargetfilepath = null;
                    string combinerTargetfilepath = null;

                    if (MapJob != null)
                    {
                        mapTargetfilepath = HomeFolder + "/temp/" + Path.GetFileName(MapJob.Filename);
                        mapper = Path.GetFileName(MapJob.Filename);
                    }
                    if (CombinerJob != null)
                    {
                        combinerTargetfilepath = HomeFolder + "/temp/" + Path.GetFileName(CombinerJob.Filename);
                        combiner = Path.GetFileName(CombinerJob.Filename);
                    }
                    if (ReduceJob != null)
                    {
                        reduceTargetfilepath = HomeFolder + "/temp/" + Path.GetFileName(ReduceJob.Filename);
                        reducer = Path.GetFileName(ReduceJob.Filename);
                    }

                    var outFile = Config.HdfsClient.GetDirectoryStatus(Output).WaitForResult();
                    if (outFile.Entries.Count() > 0)
                    {
                        var target = Output;
                        if (target.EndsWith("/")) target = target.Substring(0, target.Length - 1);
                        Config.HdfsClient.DeleteDirectory(target, true).Wait();
                    }

                    var streamJobDefintion = new StreamingMapReduceJobCreateParameters()
                    {
                        JobName = "MapReduce Studio",
                        StatusFolder = Status,
                        Input = Input,
                        Output = Output,
                        Reducer = reducer,
                        Mapper = mapper,
                        Combiner = combiner,
                    };

                    if (mapTargetfilepath != null)
                    {
                        streamJobDefintion.Files.Add(mapTargetfilepath);
                        Config.HdfsClient.CreateFile(MapJob.Filename, mapTargetfilepath).Wait();

                    }
                    if (reduceTargetfilepath != null)
                    {
                        streamJobDefintion.Files.Add(reduceTargetfilepath);
                        Config.HdfsClient.CreateFile(ReduceJob.Filename, reduceTargetfilepath).Wait();
                    }

                    if (!string.IsNullOrEmpty(Files))
                    {
                        foreach (var item in Files.Split('\n'))
                        {
                            if (!string.IsNullOrEmpty(item)) 
                            {
                                var fileName = Path.GetFileName(item);
                                var taregtFile  = HomeFolder + "/temp/" +fileName;
                                if (CopyFiles)
                                {
                                   Config.HdfsClient.CreateFile(item, taregtFile).Wait();
                                }
                                streamJobDefintion.Files.Add(taregtFile);
                            }
                        }
                    }

                    if (!string.IsNullOrEmpty(Arguments))
                    {
                        foreach (var item in Arguments.Split('\n'))
                        {
                            if (!string.IsNullOrEmpty(item)) streamJobDefintion.Arguments.Add(item);
                        }
                    }
                    if (!string.IsNullOrEmpty(Defines))
                    {
                        foreach (var item in Defines.Split('\n'))
                        {
                            var data = item.Split('=');
                            if (!string.IsNullOrEmpty(item)) streamJobDefintion.Defines.Add(data[0].Trim(), data.Length > 1 ? string.Join("=", data.Skip(1)) : "");
                        }
                    }

                    var jobResult = Config.JobClient.CreateStreamingJob(streamJobDefintion);

                    WaitForJobCompletion(jobResult, ProgressProc);
                    StdErr = ReadError();
                    Busy = false;

                };
                worker.RunWorkerAsync();
            }
        }
        private bool ProgressProc(JobDetails job)
        {
            StdErr = ReadError();
            return true;
        }

        public ICommand AddFilesCommand
        {
            get { return this.CreateCommand(AddFiles, (x) => !Busy); }
        }
        private void AddFiles()
        {
            var dialog = new OpenFileDialog();
            dialog.Filter = "*.*|*.*";
            dialog.Multiselect = true;
            if (dialog.ShowDialog() == true)
            {
                var list = new List<string>();
                if (!string.IsNullOrWhiteSpace(Files))
                {
                    foreach (var item in Files.Split('\n').Where(x => !string.IsNullOrWhiteSpace(x)))
                    {
                        list.Add(item);
                    }
                }

                foreach (var item in dialog.FileNames)
                {
                    if (list.Any(x=> string.Compare(x,item,true) == 0)) continue;
                    list.Add(item);
                }
                Files = string.Join("\n", list);
            }

        }

        protected override void Save()
        {
            JObject data = new JObject{
                { "Type", this.GetType().Name },
                { "MapJob", MapJob.Filename },
                { "ReduceJob", ReduceJob.Filename },
                { "CombinerJob", CombinerJob.Filename },
                { "Input", Input },
                { "Arguments", Arguments },
                { "Defines", Defines },
                { "Files", Files },
                { "CopyFiles", CopyFiles },
            };
            SaveData(data);
        }

        protected override void Load()
        {
            var data = LoadData();
            if (data == null) return;
            if ((string)data.Property("Type").Value != this.GetType().Name)
            {
                this.ShowMessage("Invalid selected file. " + data.Property("Type").Value);
            }
            else
            {
                this.MapJob = new MapReduceJobItem((string)data.Property("MapJob").Value);
                this.ReduceJob = new MapReduceJobItem((string)data.Property("ReduceJob").Value);
                this.CombinerJob = new MapReduceJobItem((string)data.Property("CombinerJob").Value);
                this.Input = (string)data.Property("Input").Value;
                this.Arguments = (string)data.Property("Arguments").Value;
                this.Defines = (string)data.Property("Defines").Value;
                this.Files = (string)data.Property("Files").Value;
                this.CopyFiles = (bool)data.Property("CopyFiles").Value;
            }
        }

    }
}
